package net.xdclass.func;

import com.alibaba.fastjson.JSONObject;
import net.xdclass.util.TimeUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;

public class UniqueVisitorFilterFunction extends RichFilterFunction<JSONObject> {

    private ValueState<String> lastVisitDateState = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        ValueStateDescriptor<String> visitDateStateDes = new ValueStateDescriptor<>("visitorDateState", String.class);
        //统计过期时间
//        StateTtlConfig config = StateTtlConfig.newBuilder(Time.days(1)).build();
        StateTtlConfig config = StateTtlConfig.newBuilder(Time.seconds(10)).build();
        visitDateStateDes.enableTimeToLive(config);
        this.lastVisitDateState = getRuntimeContext().getState(visitDateStateDes);
    }


    @Override
    public void close() throws Exception {
        super.close();
    }

    @Override
    public boolean filter(JSONObject jsonObject) throws Exception {
        // 获取当前访问时间
        Long visitTime = jsonObject.getLong("visitTime");
        String udid = jsonObject.getString("udid");
        //当前访问时间戳
        String currentVisitDate = TimeUtil.format(visitTime);
        //获取上次的访问时间
        String lastVisitDate = lastVisitDateState.value();

        if (StringUtils.isNotBlank(lastVisitDate) && lastVisitDate.equalsIgnoreCase(currentVisitDate)) {
            System.out.println(udid + " 已经在 " + currentVisitDate + "时间访问过");
            return false;
        }else {
            System.out.println(udid + " 在 " + currentVisitDate + "初次访问");
            lastVisitDateState.update(currentVisitDate);
            return true;
        }

    }


}
